package com.tonann.app.mqtt.singleServer;

import com.alibaba.fastjson2.JSONObject;
import com.tonann.app.domain.AppGateway;
import com.tonann.app.domain.AppSensor;
import com.tonann.app.mqtt.dao.DataBody;
import com.tonann.app.mqtt.dao.MQTTSensor;
import com.tonann.app.mqtt.dao.MessageBody;
import com.tonann.app.service.IAppGatewayService;
import com.tonann.app.service.IAppSensorService;
import com.tonann.app.task.AnalyseTask;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.handler.annotation.Header;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author Kun Chan
 * @date 2020-05-01 15:16
 */
@Configuration
public class MqttConfig {

    private MqttPahoMessageDrivenChannelAdapter adapter;

    private volatile AtomicBoolean isInit = new AtomicBoolean(false);

    @Resource
    private MqttConfig.MqttGateway mqttGateway;

    @Resource
    private IAppGatewayService gatewayService;

    @Resource
    private IAppSensorService sensorService;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private AnalyseTask analyseTask;

    private final int sameTimes = 5;

    private BigDecimal rangeTop = BigDecimal.valueOf(0.2);

    private BigDecimal rangeBottom = BigDecimal.valueOf(0.2);


    public void addListenTopics(String[] topics) {
        if (adapter == null) {
            adapter = new MqttPahoMessageDrivenChannelAdapter("client" + UUID.randomUUID().toString(),
                    mqttClientFactory(), "#");
        }
        if (adapter.isActive()) {
            for (String topic : topics) {
                adapter.addTopic(topic, 1);
            }
        }
    }

    public void addListenTopic(String topic) {
        if (adapter == null) {
            adapter = new MqttPahoMessageDrivenChannelAdapter("client" + UUID.randomUUID().toString(),
                    mqttClientFactory(), "#");
        }
        if (adapter.isActive()) {
            adapter.addTopic(topic, 1);
        }
    }


    public void removeListenerTopic(String topic) {
        if (adapter == null) {
            adapter = new MqttPahoMessageDrivenChannelAdapter("client" + UUID.randomUUID().toString(),
                    mqttClientFactory(), "register");
        }
        if (adapter.isActive()) {
            adapter.removeTopic(topic);
        }
    }

    /*****
     * 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置。
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        AppGateway appGateway = new AppGateway();
        appGateway.setGatewayType("2");
        appGateway.setStatus("0");
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        List<AppGateway> gatewayList = gatewayService.selectAppGatewayList(appGateway);
        for (AppGateway gateway : gatewayList) {
            String host = "tcp://" + gateway.getIpAddress() + ":" + gateway.getPort();
//            String host = "tcp://"  + "59.110.170.159:" + gateway.getPort();
            MqttConnectOptions options = new MqttConnectOptions();
            options.setServerURIs(new String[]{host});
            options.setAutomaticReconnect(true);
            options.setCleanSession(true);
            factory.setConnectionOptions(options);
        }
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        //防止重复触发
        if (!isInit.compareAndSet(false, true)) {
            return null;
        }
        adapter = new MqttPahoMessageDrivenChannelAdapter("client" + UUID.randomUUID().toString(),
                mqttClientFactory(), "register", "test", "#");
        Collection<String> collection = new HashSet<>();
        collection.add("#");
        List<AppSensor> appSensorList = sensorService.selectAppSensorListByGatewayType("2");
        for (AppSensor sensor : appSensorList) {
            String tag = sensor.getMqttDeviceChannel() + "." + sensor.getMqttDeviceName() + "." + sensor.getMqttDeviceTag();
            MQTTSensor.getInstance().map.put(tag, sensor);
            if (com.tonann.common.utils.StringUtils.isNotEmpty(sensor.getMqttTopic()) && !collection.contains(sensor.getMqttTopic())) {
                collection.add(sensor.getMqttTopic());
                MQTTSensor.getInstance().topicSet.add(sensor.getMqttTopic());
                adapter.addTopic(sensor.getMqttTopic(), 1);
            }
        }
        adapter.setCompletionTimeout(50000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setManualAcks(false);
//        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            String payload = message.getPayload().toString();
            String topic = Objects.requireNonNull(message.getHeaders().get("mqtt_receivedTopic")).toString();
//            String clientId = message.getHeaders().get("id").toString();
//            String mqttDuplicate = message.getHeaders().get("mqtt_duplicate").toString();
//            String qos = message.getHeaders().get("mqtt_receivedQos").toString();
//            Long timestamp = message.getHeaders().getTimestamp().longValue();
//            System.out.println(topic + "======" + payload);
            new Thread(() -> {
                // 根据topic分别进行消息处理。
                try {
                    MessageBody messageBody = JSONObject.parseObject(payload, MessageBody.class);
                    List<AppSensor> updateList = new ArrayList<>();
                    for (DataBody dataBody : messageBody.getValues()) {
                        String tagId = dataBody.getId();
                        String v = dataBody.getValue();
                        if (v.trim().isEmpty()) {
                            continue;
                        } else if (v.trim().equals("true")) {
                            v = "1";
                        } else if (v.trim().equals("false")) {
                            v = "0";
                        }
                        if (MQTTSensor.getInstance().count == (Long.MAX_VALUE - 10L)) {
                            MQTTSensor.getInstance().count = 0L;
                        }
                        MQTTSensor.getInstance().count++;
                        AppSensor sensor = MQTTSensor.getInstance().map.get(tagId);
                        if (sensor != null) {
                            String oldValue = sensor.getValue();
                            if (v.contains(".")) {
                                v = format(v);
                            }
                            float result = new BigDecimal(v).setScale(2, RoundingMode.DOWN).floatValue();
                            long magnification = sensor.getMagnification() == null ? 1L : sensor.getMagnification();
                            float r = result / magnification;
                            BigDecimal bigDecimal = new BigDecimal(r + "");
                            bigDecimal = bigDecimal.setScale(2, RoundingMode.DOWN);
                            String updateValue = bigDecimal.floatValue() + "";
                            sensor.setValue(updateValue);
                            String userName = "modbus";
                            sensor.setUpdateBy(userName);
                            sensor.setUpdateTime(new Timestamp(new Date().getTime()));
                            updateList.add(sensor);
                            int valueCount = sensor.getValueCount();
                            try {
                                if (valueCount == 0 && !sensor.getAlarming()) {
                                    int res = analyseTask.faultAnalyse(sensor);
                                    if (res == 1) { // 如果触发告警
                                        sensor.setValueCount(1);
                                        sensor.setAlarming(true);
                                    } else {
                                        sensor.setValueCount(0);
                                    }
                                } else if (sensor.getAlarming()) {
                                    //如果相同的值出现规定次数以上，就重新赋值，可以进行告警分析
                                    if (valueCount >= sameTimes) {
                                        sensor.setValueCount(0);
                                        sensor.setAlarming(false);
                                    } else {
                                        if (oldValue != null && !oldValue.isEmpty()) {
                                            BigDecimal oldValueBigDecimal = new BigDecimal(oldValue);
                                            rangeTop = oldValueBigDecimal.add(rangeTop);
                                            rangeBottom = oldValueBigDecimal.subtract(rangeBottom);
                                            //比较新值是否在可以波动的范围内
                                            if (bigDecimal.compareTo(rangeTop) <= 0 && bigDecimal.compareTo(rangeBottom) >= 0) {
                                                sensor.setValueCount(sensor.getValueCount() + 1);
                                                sensor.setValue(oldValue);
                                            }
                                        } else {
                                            sensor.setValueCount(sensor.getValueCount() + 1);
                                        }
                                    }
                                }
                                MQTTSensor.getInstance().map.put(tagId, sensor);
                            } catch (Exception e) {
//                                e.printStackTrace();
                            }
                        }
                    }
                    analyseTask.analyseList(updateList);
                } catch (Exception e) {
//                    e.printStackTrace();
//                    System.out.println("有错误" + payload);
                }
            }).start();
        };
    }

    public static String format(String input) {
        Pattern pattern = Pattern.compile("\\d+\\.\\d{2}");
        Matcher matcher = pattern.matcher(input);
        if (matcher.find()) {
            return matcher.group();
        }
        return input;
    }

    // 发送消息

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /*****
     * 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler outbound() {
        // 在这里进行mqttOutboundChannel的相关设置
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler("publishClient", mqttClientFactory());
        messageHandler.setAsync(true); //如果设置成true，发送消息时将不会阻塞。
        messageHandler.setDefaultTopic("testTopic");
        return messageHandler;
    }


    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MqttGateway {
        // 定义重载方法，用于消息发送
        void sendToMqtt(String payload);

        // 指定topic进行消息发送
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.RETAINED) Boolean retained, String payload);

        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.RETAINED) Boolean retained, @Header(MqttHeaders.QOS) int qos, String payload);
    }

}
